package com.niit.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Date:2025/5/22
 * Author：Ys
 * Description:
 */
object StreamingWindow06 {

  def main(args: Array[String]): Unit = {
    val ssc = new StreamingContext(new SparkConf().setMaster("local[*]").setAppName("StreamingTransform04"), Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    val wordOne: DStream[(String, Int)] = lines.flatMap(_.split(" ")).map((_, 1))

    /*
     窗口的范围必须是采集周期的整数倍
     第一个参数：窗口的时长 （一次采集可以采集几个周期的数据） 6s
     第二个参数：滑块步长 （窗口移动几个周期） 6s
     */
    val winDS: DStream[(String, Int)] = wordOne.window(Seconds(6), Seconds(6))

    val wordCount: DStream[(String, Int)] = winDS.reduceByKey(_ + _)

    wordCount.print()

    ssc.start()
    ssc.awaitTermination()

  }

}
